package core

import (
	"bytes"
	"fmt"
	"net"
	"sync"
)

type LoginClient struct {
	ClientID uint64
	Conn     net.Conn
}
type Broker struct {
	Clients         map[net.Conn]*Client
	IDS             map[uint64]net.Conn
	TopicMsgChannel chan *TopicMessage // 用于发布消息的通道
	P2PMsgChannel   chan *P2PMessage

	//检测是否有重复登陆
	CheckLogin chan *LoginClient
}

var (
	broker_instance *Broker
	once_instance   sync.Once
)

func GetBroker() *Broker {
	once_instance.Do(func() {
		broker_instance = NewBroker()
	})
	return broker_instance
}

func NewBroker() *Broker {
	return &Broker{
		Clients:         make(map[net.Conn]*Client),
		IDS:             make(map[uint64]net.Conn),
		TopicMsgChannel: make(chan *TopicMessage, 1000), // 初始化发布消息通道
		P2PMsgChannel:   make(chan *P2PMessage, 1000),
		CheckLogin:      make(chan *LoginClient, 1000),
	}
}
func (b *Broker) ProcessData(ref interface{}, conn net.Conn, state int, err error, data []byte, datasize uint32) {
	switch state {
	case TCP_STATE_CONNECTED:
		b.AddClient(conn)
	case TCP_STATE_DISCONNECTED:
		GetLog().WriteLog(Log_Warn, "Client DisConnect,Error:", err)
		b.RemoveClient(conn)
	case TCP_STATE_DATA:
		if client, exists := b.GetClient(conn); exists {
			if err := client.ProcessData(data); err != nil {
				GetLog().WriteLog(Log_Error, "Process Data Error:", err.Error())
			}
		}
	case TCP_STATE_ERROR:
		if _, exists := b.GetClient(conn); exists {
			GetLog().WriteLog(Log_Warn, "Client DisConnect,Error:", err)
			b.RemoveClient(conn)
		}
	}
}
func (b *Broker) AddClient(conn net.Conn) {
	client := NewClient(conn)
	b.Clients[conn] = client
	GetLog().WriteLog(Log_Debug, "Add Client:", conn.RemoteAddr().String())
}
func (b *Broker) RemoveClient(conn net.Conn) {
	b.Clients[conn].Conn.Close()
	delete(b.Clients, conn)
	GetLog().WriteLog(Log_Debug, "Remove Client:", conn.RemoteAddr().String())
}
func (b *Broker) GetClient(conn net.Conn) (*Client, bool) {
	client, exists := b.Clients[conn]
	return client, exists
}

func (b *Broker) ProcessTopicMsg(msg *TopicMessage) {
	topics := GetTopicMng().SelectTopic(msg.TopicName)
	if topics != nil {
		for _, client := range topics.Clients {
			client.SendTopicMsg(msg)
		}
	}
}
func (b *Broker) ProcessP2PMsg(msg *P2PMessage) {
	for _, client := range b.Clients {
		if client.ClientID == msg.TargetClientID {
			buf, err := EncodeP2PMessage(msg)
			if err != nil {
				return
			}
			client.SendBytes(buf)
		}
	}
}
func (b *Broker) PostPublishMsg(msg *TopicMessage) error {
	if msg == nil {
		return fmt.Errorf("nil PublishMessage cannot be posted")
	}
	// 如果通道已满，返回错误
	b.TopicMsgChannel <- msg
	return nil
}
func (b *Broker) StartProcessTask() {
	go func() {
		for msg := range b.TopicMsgChannel {
			b.ProcessTopicMsg(msg)
		}
	}()

	go func() {
		for msg := range b.P2PMsgChannel {
			b.ProcessP2PMsg(msg)
		}
	}()

	go func() {
		for loginInfo := range b.CheckLogin {
			for _, client := range b.Clients {
				if client.ClientID == loginInfo.ClientID && client.Conn != loginInfo.Conn {
					//找到旧登录客户端，关闭它
					//通知客户端，你断开了，客户端不再重新连接
					bufs := bytes.NewBuffer(nil)
					bufs.WriteByte(MQ_DisConnect)
					WriteString(bufs, "New Client Login,This Client Close")
					client.SendBytes(bufs.Bytes())
					//断开连接
					client.Close()
				}
			}
		}
	}()
}
